package com.yzh.dxpro.p9.taskNine;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

import com.yzh.dxpro.p1.clear.TProperties;
	
	
public class NineMapper extends Mapper<LongWritable,Text,Text,Text>{
	private Text okey = new Text();
	private Text ovalue = new Text();
	private Map<String,String> typeData = new HashMap<String,String>();
	private Map<String,Integer> userType = new HashMap<String,Integer>();
	private String typeKey;
	protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, Text>.Context context)
	throws IOException, InterruptedException {
	    String[] values = value.toString().split(TProperties.getValue("fileoutsplit"));
		if(typeData.containsKey(values[0])) {
			typeKey = typeData.get(values[0])+","+values[1];
			if(userType.containsKey(typeKey)) {
				userType.put(typeKey,userType.get(typeKey)+Integer.parseInt(values[2]));
				
			}else {
				userType.put(typeKey,Integer.parseInt(values[2]));
			}
		}
	
	}


protected void setup(Mapper<LongWritable,Text,Text,Text>.Context context) 
	throws IOException,InterruptedException{
	 Path file = new Path(context.getCacheFiles()[0].getPath());
	 BufferedReader reader = new BufferedReader(new FileReader(file.toString()));
	 String str = null;
	 try {
		 while((str=reader.readLine())!=null) {
			 String[] splits = str.split(TProperties.getValue("fileoutsplit"));
			 typeData.put(splits[1],splits[0]);
		 	}
		 }catch(Exception e) {
			 e.printStackTrace();
		 }finally {
			 reader.close();
		 }
	 }
protected void cleanup(Mapper<LongWritable,Text,Text,Text>.Context context) 
	throws IOException,InterruptedException{
	 for(String key:userType.keySet()) {
		 okey = new Text(key.split(",",-1)[0]);
		 ovalue = new Text(key.split(",",-1)[1]+TProperties.getValue("fileoutsplit")+userType.get(key));
		 context.write(okey,ovalue);
	 }
	}
}